package com.danbay.framework.kafka;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.Serializer;
import org.eclipse.californium.core.coap.Request;
import org.eclipse.californium.core.coap.Response;
import org.eclipse.californium.core.network.serialization.DataSerializer;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.Map;

/**
 * @author SJG
 *         2017/11/13.
 */
@Slf4j
public class AccessPackageSerializer<V> implements Serializer<AccessPackage<V>> {

    private String encoding = "utf8";
    private ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        String propertyName = isKey ? "key.serializer.encoding" : "value.serializer.encoding";
        Object encodingValue = configs.get(propertyName);
        if (encodingValue == null) {
            encodingValue = configs.get("serializer.encoding");
        }

        if (encodingValue != null && encodingValue instanceof String) {
            this.encoding = (String) encodingValue;
        }

    }

    @Override
    public byte[] serialize(String topic, AccessPackage<V> accessPackage) {
        byte[] contentBytes = new byte[0];
        String contentString = "";
        Object originData = accessPackage.getData();
        //如果是CoAP格式的请求，直接转换为CoAP请求内容
        if (originData instanceof Request) {
            Request request = (Request) originData;
            contentString = request.toString();
            if (log.isDebugEnabled()) {
                log.debug("serialize CoAP package before send:{}", contentString);
            }
            contentBytes = new DataSerializer().serializeRequest(request);
        } else if (originData instanceof Response) {
            //CoAP格式的响应内容
            Response response = (Response) originData;
            contentString = response.toString();
            if (log.isDebugEnabled()) {
                log.debug("serialize CoAP package before send:{}", contentString);
            }
            contentBytes = new DataSerializer().serializeResponse(response);
        } else if (originData != null) {
            try {
                //如果是非CoAp格式的数据，直接转化为json字符串
                if (originData.getClass().getName().startsWith("java.lang.")) {
                    //lang包中的原生类型
                    contentString = originData.toString();
                    contentBytes = originData.toString().getBytes(encoding);
                } else {
                    //其它类型
                    contentString = new ObjectMapper().writeValueAsString(originData);
                    contentBytes = contentString.getBytes(encoding);
                }
            } catch (JsonProcessingException e) {
                log.error("消息体json序列化失败", e);
                return new byte[0];
            } catch (UnsupportedEncodingException e) {
                log.error("消息体byte序列化失败", e);
            }
        }
        try {
            AccessHeader header = accessPackage.getHeader();
            //序列化消息头
            String headerJson = objectMapper.writeValueAsString(header);
            byte[] headerBytes = headerJson.getBytes(encoding);
            ByteBuffer byteBuffer;
            byteBuffer = ByteBuffer.allocate(4 + headerBytes.length + contentBytes.length);
            //写入消息头
            byteBuffer.putInt(headerBytes.length);
            byteBuffer.put(headerBytes);
            //消息体内容
            byteBuffer.put(contentBytes);
            byte[] msgBytes = byteBuffer.array();
            byteBuffer.clear();

            if (log.isDebugEnabled()) {
                log.debug("serialize kafka package before send,header:{},data:{}",
                        headerJson, contentString);
            }
            return msgBytes;
        } catch (UnsupportedEncodingException e) {
            log.error("消息source byte序列化失败", e);
        } catch (JsonProcessingException e) {
            log.error("消息头序列化失败", e);
        }
        return new byte[0];
    }

    @Override
    public void close() {

    }
}
